package com.kancy.delay.queue.redis.service.impl;

import com.kancy.delay.queue.redis.config.DelayQueueRedisTemplate;
import com.kancy.delay.queue.redis.config.RedisScriptConfig;
import com.kancy.delay.queue.redis.exception.DelayMessageSendException;
import com.kancy.delay.queue.redis.key.RedisKeyResolver;
import com.kancy.delay.queue.redis.message.DelayMessage;
import com.kancy.delay.queue.redis.properties.DelayQueueProperties;
import com.kancy.delay.queue.redis.service.DelayQueueClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * <p>
 * DelayQueueClientImpl
 * <p>
 *
 * @author: kancy
 * @date: 2020/3/4 20:38
 **/
public class DelayQueueClientImpl implements DelayQueueClient {
    private final Logger log = LoggerFactory.getLogger(DelayQueueClientImpl.class);

    private final DelayQueueRedisTemplate delayQueueRedisTemplate;
    private final RedisKeyResolver redisKeyResolver;
    private final DelayQueueProperties delayQueueProperties;

    private final Map<String, Boolean> topicWarnMap = new HashMap<>();

    public DelayQueueClientImpl(DelayQueueRedisTemplate delayQueueRedisTemplate, RedisKeyResolver redisKeyResolver,
                                DelayQueueProperties delayQueueProperties) {
        this.delayQueueRedisTemplate = delayQueueRedisTemplate;
        this.redisKeyResolver = redisKeyResolver;
        this.delayQueueProperties = delayQueueProperties;
    }

    /**
     * 发送消息
     *
     * @param delayMessage
     * @throws DelayMessageSendException
     * @return
     */
    @Override
    public boolean sendMessage(DelayMessage delayMessage) {

        // 1.检查消息是否合法
        checkDelayMessage(delayMessage);

        // 2.保存消息到redis
        try {
            saveDelayMessage(delayMessage);
        } catch (Exception e) {
            throw new DelayMessageSendException(String.format("Send delayMessage [%s] to [topic:%s] is failed.",
                    delayMessage.getId(), delayMessage.getTopic()), e);
        }
        return true;
    }

    /**
     * 保存消息到redis, 执行lua脚本保证事务一致性
     * @param delayMessage
     */
    private void saveDelayMessage(DelayMessage delayMessage) {
        // 保存traceId，便于在分布式环境中请求链路完整性
        delayMessage.setTraceId(MDC.get(DelayMessage.TRACE_ID_KEY));

        List<String> keys = new ArrayList<>();
        keys.add(redisKeyResolver.hashTableKey(delayMessage.getTopic()));
        keys.add(redisKeyResolver.key(delayMessage.getId()));
        keys.add(redisKeyResolver.bucketKey(delayMessage.getTopic()));
        delayQueueRedisTemplate.execute(
                RedisScriptConfig.getDelayMessageSaveRedisScript(), keys,
                delayMessage, delayMessage.getId(), getDelayScore(delayMessage.getDelay()));
    }

    /**
     * 检查消息
     * @param delayMessage
     */
    private void checkDelayMessage(DelayMessage delayMessage) {
        Assert.notNull(delayMessage,"delayMessage is null.");
        Assert.hasText(delayMessage.getId(),"delayMessage id is empty.");
        Assert.notNull(delayMessage.getDelay(),"delayMessage delay time no set.");

        String topic = delayMessage.getTopic();
        Assert.hasText(topic,"delayMessage topic is empty.");
        Assert.isTrue(delayQueueProperties.getTopics().keySet().contains(delayMessage.getTopic()),
                String.format("delay queue not register topic : %s", topic));

        // 延时时间设置警告，比扫描时间小则提示一次
        delayTimeWaring(delayMessage, topic);

    }

    /**
     * 延时时间设置警告
     * @param delayMessage
     * @param topic
     */
    private void delayTimeWaring(DelayMessage delayMessage, String topic) {
        // 延时时间设置警告，比扫描时间小则提示
        if (!topicWarnMap.containsKey(topic)){
            topicWarnMap.put(topic, Boolean.FALSE);
            long delayMills = delayMessage.getDelay();
            long scanIntervalMillis = delayQueueProperties.getServer().getScanInterval().toMillis();
            if (scanIntervalMillis > delayMills){
                log.warn("The scanning interval [{}] is longer than the delay time [{}] of the topic [{}] message. There may be errors in the delay mechanism.",
                        scanIntervalMillis, delayMills, delayMessage.getTopic());
                topicWarnMap.put(topic, Boolean.TRUE);
            }
        }
    }

    /**
     * z-set score
     * @param delay
     * @return
     */
    private Long getDelayScore(Long delay) {
        return System.currentTimeMillis() + delay;
    }
}
